package com.smart.thrift.discoveryandbalance.discovery.zookeeper;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.smart.thrift.discoveryandbalance.discovery.ThriftServerNodeList;
import lombok.SneakyThrows;
import org.apache.curator.framework.CuratorFramework;

import java.util.*;
import java.util.stream.Collectors;

import static com.smart.thrift.discoveryandbalance.discovery.zookeeper.ZookeeperRegistry.ZOOKEEPER_NODE_PREFIX;

public class ThriftZookeeperServerNodeList extends ThriftServerNodeList<ThriftZookeeperServerNode> {


    private final ZookeeperRegistry zookeeperRegistry;

    private final CuratorFramework zkClient;

    private static volatile ThriftZookeeperServerNodeList serverNodeList = null;

    public static ThriftZookeeperServerNodeList singleton(ZookeeperRegistry zookeeperRegistry) {
        if (serverNodeList == null) {
            synchronized (ThriftZookeeperServerNodeList.class) {
                if (serverNodeList == null) {
                    serverNodeList = new ThriftZookeeperServerNodeList(zookeeperRegistry);
                }
            }
        }
        return serverNodeList;
    }

    private ThriftZookeeperServerNodeList(ZookeeperRegistry zookeeperRegistry) {
        this.zookeeperRegistry = zookeeperRegistry;
        this.zkClient = this.zookeeperRegistry.getClient();
        refreshThriftServers();
    }

    @Override
    public List<ThriftZookeeperServerNode> getThriftServer(String serviceName) {
        if (!this.serverNodeMap.isEmpty() && (this.serverNodeMap.containsKey(serviceName))) {
            LinkedHashSet<ThriftZookeeperServerNode> serverNodeSet = this.serverNodeMap.get(serviceName);
            if (!serverNodeSet.isEmpty()) {
                return Lists.newArrayList(serverNodeSet);
            }
        }

        return refreshThriftServer(serviceName);
    }

    @Override
    @SneakyThrows
    public List<ThriftZookeeperServerNode> refreshThriftServer(String serviceName) {
        List<String> childNodePaths = zkClient.getChildren().watched().forPath(ZOOKEEPER_NODE_PREFIX + "/" + serviceName);
        List<String> childNodeFullPaths = childNodePaths.stream().map(path -> serviceName + "/" + path).collect(Collectors.toList());
        Set<ThriftZookeeperServerNode> thriftZookeeperServerNodes = zookeeperRegistry.serviceNodes(childNodeFullPaths);
        if (!thriftZookeeperServerNodes.isEmpty()) {
            this.serverNodeMap.put(serviceName, Sets.newLinkedHashSet(thriftZookeeperServerNodes));
        }
        return new ArrayList<>(thriftZookeeperServerNodes);
    }

    @Override
    public Map<String, LinkedHashSet<ThriftZookeeperServerNode>> getThriftServers() {
        if (!this.serverNodeMap.isEmpty()) {
            return ImmutableMap.copyOf(this.serverNodeMap);
        }

        return refreshThriftServers();
    }

    @Override
    @SneakyThrows
    public Map<String, LinkedHashSet<ThriftZookeeperServerNode>> refreshThriftServers() {
        List<String> serviceNames = zkClient.getChildren().watched().forPath(ZOOKEEPER_NODE_PREFIX);
        for (String serviceName : serviceNames) {
            List<String> childNodePaths = zkClient.getChildren().watched().forPath(ZOOKEEPER_NODE_PREFIX+ "/" + serviceName);
            List<String> childNodeFullPaths = childNodePaths.stream().map(path -> serviceName + "/" + path).collect(Collectors.toList());
            Set<ThriftZookeeperServerNode> thriftZookeeperServerNodes = zookeeperRegistry.serviceNodes(childNodeFullPaths);
            this.serverNodeMap.put(serviceName, Sets.newLinkedHashSet(thriftZookeeperServerNodes));
        }
        return ImmutableMap.copyOf(this.serverNodeMap);
    }

}
